package com.alpaak.common.utils;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.*;

public class KafkaUtils {
	
	/**
	 * 接收数据
	 */
	public static List<String> ConsumerReceive(String topic){
		Properties properties = new Properties();  
		properties.put("zookeeper.connect", "192.168.116.144:2181,"
				+ "192.168.116.146:2181,192.168.116.147:2181");//声明zk  
		properties.put("group.id", "group2xxx1212");// 必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据  
		properties.put("auto.offset.reset", "smallest");
//		properties.put("zookeeper.session.timeout.ms", "400");
//		properties.put("zookeeper.sync.time.ms", "200");
//		properties.put("auto.commit.interval.ms", "1000");
//		properties.put("serializer.class", "kafka.serializer.StringEncoder");
		ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
	
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
		topicCountMap.put(topic, 1); // 一次从主题中获取一个数据  
		Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
		KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据  
		ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
		
		List<String> json = new ArrayList<String>();
		while(iterator.hasNext()){
			String data = new String(iterator.next().message());
			System.out.println("receive：" + data);
			json.add(data);
		}
		return json;
	}
	
	/**
	 * 发送数据
	 * @param topic
	 * @param value
	 */
	public static void ProducerSend(String topic,String value){
		Properties props = new Properties();
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		props.put("metadata.broker.list", "192.168.116.148:9092");
		Producer<Integer, String> producer = new Producer<Integer, String>(new ProducerConfig(props));
		producer.send(new KeyedMessage<Integer, String>(topic,value));
		producer.close();
	}
	
	
	
}
